package com.zlx.rulemk.datagen;

import com.alibaba.fastjson.JSON;
import com.typesafe.config.ConfigFactory;
import com.zlx.rulemk.marketing.beans.LogBean;
import com.zlx.rulemk.marketing.constant.ConfigNames;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-06
 * @Description: 模拟一条行为日志
 * @Version: 1.0
 */
public class _05_ActionLogGenOne {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", ConfigFactory.load().getString(ConfigNames.KAFKA_BOOTSTRAP_SERVERS));
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);

        LogBean logBean = new LogBean();
        logBean.setDeviceId("000001");
        logBean.setEventId("K");
        Map<String, String> ps = new HashMap();
        ps.put("p1", "v2");
        logBean.setProperties(ps);
        logBean.setTimeStamp(System.currentTimeMillis());

        String log = JSON.toJSONString(logBean);
        ProducerRecord<String, String> record = new ProducerRecord<>(ConfigFactory.load().getString(ConfigNames.KAFKA_ACTION_DETAIL_TOPIC), log);
        kafkaProducer.send(record);
        kafkaProducer.flush();

        System.out.println("发送模拟的日志数据成功");

    }
}


